"""
Case Type   : 基础功能
Case Name   : create subscription copy_data为false时，refresh copy_data
Create At   : 2022/11/09
Owner       : opentestcase026
Description :
    1.两个集群均创建表1
    2.发布端集群在表1内插入数据
    3.创建发布端
    4.创建订阅端
      4.1创建订阅端设置copy_data值为空
      4.2创建订阅端设置copy_data值为‘false’
    5.查询订阅端表1数据
    6.两个集群均创建表2
    7.发布端在表1和表2内插入数据
    8.发布端添加表2,更新订阅端copy_data为true
    9.查询表1与表2数据
    10.两个集群均创建表3
    11.发布端插入数据
    12.发布端添加表3,更新订阅端copy_data为false
    13.查询表1,2,3内数据
    14.更新订阅端copy_data为true
    15.查询表3数据
    16.发布端在表3内插入数据
    17.查询表3数据
    18.refresh更新copy_data为非法值
    19.删除订阅
    20.删除发布
    21.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.部分测试点报错
      4.1合理报错
      4.2成功
    5.数据未同步
    6.成功
    7.成功
    8.成功
    9.表1同步1条数据，表2同步2条数据
    10.成功
    11.成功
    12.成功
    13.新数据在表1,2内同步,在表3内未同步
    14.成功
    15.表3内数据未同步
    16.成功
    17.新数据在表3内同步
    18.合理报错
    19.成功
    20.成功
    21.成功
History     :
"""
import unittest
import os

from yat.test import macro
from yat.test import Node

from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != primary_sh.get_node_num(), '非1+2环境不执行')
class PubSubClass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} start-----")
        self.pri_user_pub = Node(node='PrimaryDbUser')
        self.pri_user_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.com_sh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.tb_name1 = 'tb_pub_sub_case246_1'
        self.tb_name2 = 'tb_pub_sub_case246_2'
        self.tb_name3 = 'tb_pub_sub_case246_3'
        self.sub_name = "sub_case246"
        self.pub_name = "pub_case246"
        self.port = int(self.pri_user_pub.db_port) + 1
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_user_pub.db_user} ' \
            f'-W {self.pri_user_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_user_sub.db_user} ' \
            f'-W {self.pri_user_sub.db_password}'
        text = '----修改pg_hba expect:成功----'
        self.log.info(text)
        self.log.info('----发布端状态----')
        status = primary_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        self.log.info('----订阅端状态----')
        status = \
            self.com_sh_sub.get_db_cluster_status(
                'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        guc_res = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_user_sub.db_user} '
            f'{self.pri_user_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = primary_sh.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)
        guc_res = self.com_sh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_user_pub.db_user} '
            f'{self.pri_user_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:两个集群均创建表1 expect:成功--'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name1};
                create table {self.tb_name1}(i int,t text);'''
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)

        text = '--step2:发布端在表1内插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name1} values(1,'first'),(2,'second');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step3:创建发布端;expect:成功--'
        self.log.info(text)
        sql = f"drop publication if exists {self.pub_name};" \
              f"create publication {self.pub_name} " \
              f"for table {self.tb_name1};"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.drop_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '--step4:创建订阅端;expect:部分测试点合理报错--'
        self.log.info(text)
        result = self.com_sh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        text = '--step4.1:创建订阅端,copy_data为空值;expect:失败--'
        sql1 = f"drop subscription if exists {self.sub_name};" \
               f"create subscription {self.sub_name} connection " \
               f"'host={self.pri_user_pub.db_host} " \
               f"port={self.port} " \
               f"user={self.pri_user_pub.db_user} " \
               f"dbname={self.pri_user_pub.db_name} " \
               f"password={self.pri_user_pub.ssh_password}' " \
               f"publication {self.pub_name} with (copy_data='');"
        result = self.com_sh_sub.execut_db_sql(sql1, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.drop_sub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn('ERROR', result, '执行失败:' + text)
        text = '--step4.2:创建订阅端,copy_data值为false;expect:成功--'
        sql2 = f"drop subscription if exists {self.sub_name};" \
               f"create subscription {self.sub_name} connection " \
               f"'host={self.pri_user_pub.db_host} " \
               f"port={self.port} " \
               f"user={self.pri_user_pub.db_user} " \
               f"dbname={self.pri_user_pub.db_name} " \
               f"password={self.pri_user_pub.ssh_password}' " \
               f"publication {self.pub_name} with (copy_data=false);"
        result = self.com_sh_sub.execut_db_sql(sql2, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.drop_sub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)

        text = '--step5:查询订阅端表1数据 expect:数据未同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 rows'), 1, '执行失败' + text)

        text = '--step6:两个集群均创建表2 expect:成功--'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name2};
                create table {self.tb_name2}(i int,t text);'''
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)

        text = '--step7:发布端插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name2} values(1,'first'),(2,'second');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)
        sql = f"insert into {self.tb_name1} values(3,'first');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step8:发布端添加表2,更新订阅端copy_data为true expect:成功--'
        self.log.info(text)
        sql = f"alter publication {self.pub_name} add table {self.tb_name2};"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.alter_pub_succ_msg, result, '执行失败' +
                      text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=true);"\
              f"select pg_sleep(2);" \
              f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=true);" \
              f"select pg_sleep(2);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.alter_sub_succ_msg, result, '执行失败' +
                      text)

        text = '--step9:查询表1与表2数据 expect:表1同步1条数据,表2同步2条数据--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 1, '执行失败' + text)
        self.assertIn('3 | first', result, '执行失败' + text)
        sql = f"select * from {self.tb_name2};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('2 rows'), 1, '执行失败' + text)

        text = '--step10:两个集群均创建表3 expect:成功--'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name3};
                  create table {self.tb_name3}(i int,t text);'''
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)

        text = '--step11:发布端插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name3} values(1,'first'),(2,'second');" \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name1} values(4,'step11');" \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name2} values(4,'step11');"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step12:发布端添加表3,更新订阅端copy_data为false expect:成功--'
        self.log.info(text)
        sql = f"alter publication {self.pub_name} add table {self.tb_name3};"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.alter_pub_succ_msg, result, '执行失败' +
                      text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=false);" \
              f"select pg_sleep(2)" \
              f"alter subscription {self.sub_name}" \
              f"refresh publication with (copy_data=false);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.alter_sub_succ_msg, result, '执行失败' +
                      text)

        text = '--step13:查询表1,2,3内数据 expect:新数据在表1,2内同步,在表3内未同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('2 rows'), 1, '执行失败' + text)
        self.assertIn('4 | step11', result, '执行失败' + text)
        sql = f"select * from {self.tb_name2};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('3 rows'), 1, '执行失败' + text)
        self.assertIn('4 | step11', result, '执行失败' + text)
        sql = f"select * from {self.tb_name3};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 rows'), 1, '执行失败' + text)

        text = '--step14:更新订阅端copy_data为true expect:成功--'
        self.log.info(text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=true);" \
              f"select pg_sleep(2)" \
              f"alter subscription {self.sub_name}" \
              f"refresh publication with (copy_data=true);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.alter_sub_succ_msg, result, '执行失败' +
                      text)

        text = '--step15:查询表3数据 expect:表3数据未同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name3};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 rows'), 1, '执行失败' + text)

        text = '--step16:发布端在表3内插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name3} values(5,'step16');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step17:查询表3数据 expect:新数据已同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name3};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 1, '执行失败' + text)
        self.assertIn('5 | step16', result, '执行失败' + text)

        text = '--step18:refresh更新copy_data为非法值 expect:合理报错--'
        self.log.info(text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=none);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('ERROR', result, '执行失败' +
                      text)

    def tearDown(self):
        self.log.info('------------this is tearDown-------------')
        text = '--step19:删除订阅 expect:成功--'
        self.log.info(text)
        sql = f"drop subscription  {self.sub_name};"
        drop_sub_result = self.com_sh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)

        text = '--step20:删除发布 expect:成功--'
        self.log.info(text)
        sql = f"drop publication {self.pub_name};"
        drop_pub_result = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text = '--step21:清理环境,删除表 expect:成功--'
        self.log.info(text)
        sql = f"drop table {self.tb_name1},{self.tb_name2},{self.tb_name3};"
        result1 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result1)
        result2 = primary_sh.execut_db_sql(sql,
                                           sql_type=self.user_param_pub)
        self.log.info(result2)
        guc_res2 = primary_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(guc_res2)
        guc_res1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_user_sub.db_user} '
            f'{self.pri_user_sub.db_host}/32')
        self.log.info(guc_res1)
        guc_res = self.com_sh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_user_pub.db_user} '
            f'{self.pri_user_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        result = primary_sh.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result \
              or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertTrue(guc_res, '执行失败:' + text)
        self.assertTrue(guc_res1, '执行失败:' + text)
        self.assertTrue(guc_res2, '执行失败:' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} end-----")